package com.guo.rabbitmq.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author gonghuaiguo
 */
@Component
@Slf4j
public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME = "queue.order";

    /**
     *  1.写法一：监听 + 在rabbitmq中创建交换机、队列及绑定关系
     * @RabbitListener(bindings = @QueueBinding(
     *         value = @Queue(value = QUEUE_NAME,durable = "true"),
     *         exchange = @Exchange(value = EXCHANGE_DIRECT),
     *         key = {ROUTING_KEY}
     * ))
     * 2. 写法二：监听
     * @RabbitListener(queues = {QUEUE_NAME})
     *
     *  deliveryTag(交付标签机制)：存入队列的每条消息都有一个唯一交付标签
     *
     *  消息确认的相关参数
     *  multiple: 为true时批量处理消息，为false时单条处理消息
     *
     * @param dataString
     * @param message
     * @param channel
     */
//    @RabbitListener(queues = {QUEUE_NAME})
    public void processMessage(String dataString, Message message, Channel channel) throws IOException {

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            log.info("消费端接受到了消息：" + dataString);
            // 返回ACK
            channel.basicAck(deliveryTag,false);
        } catch (Exception e) {

            // 返回NACK
            // requeue
            //      取值为true 则将这个消息重新放回队列，broker 会重新投递这个消息
            //      取值为false 不重新放回队列， broker 放弃投递这个消息
            Boolean requeue = message.getMessageProperties().getRedelivered()?false:true;

            channel.basicNack(deliveryTag,false,requeue);
        }
    }

//    @RabbitListener(queues = {QUEUE_NAME})
    public void processMessagePrefetch(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        // 在配置文件设置prefetch
        log.info("削峰限流 prefetch ：" +dataString);
        Thread.sleep(1000);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
    }

    public static final String QUEUE_NORMAL_NAME = "queue.normal.video";
    public static final String QUEUE_DEAD_NAME = "queue.dead.letter.video";

    /**
     * 监听正常队列
     * @param dataString
     * @param message
     * @param channel
     * @throws IOException
     * @throws InterruptedException
     */
//    @RabbitListener(queues = {QUEUE_NORMAL_NAME})
    public void processMessageNormal(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        // 在配置文件设置prefetch
        log.info("[normal]消息正常接收 但我拒绝：" +dataString);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicReject(deliveryTag,false);
    }

    /**
     * 监听死信队列
     * @param dataString
     * @param message
     * @param channel
     * @throws IOException
     * @throws InterruptedException
     */
//    @RabbitListener(queues = {QUEUE_DEAD_NAME})
    public void processMessageDead(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
        // 在配置文件设置prefetch
        log.info("[dead letter] dataString =" +dataString);
        log.info("[dead letter] 我是死信监听方法，监听到了死信消息" );
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        channel.basicAck(deliveryTag,false);
    }

    public static final String QUEUE_FED_NAME = "fed.queue.demo";

//    @RabbitListener(queues = {QUEUE_FED_NAME})
    public void processMessageFed(String dataString,Message message,Channel channel) throws IOException {
        log.info("[federation queue] 获取消息" +dataString);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }
}
